package cn.oowoo.rabbitmq.dlx;

import cn.oowoo.rabbitmq.config.RabbitConfig;
import cn.oowoo.rabbitmq.config.RabbitExchangeTypeEnum;
import com.rabbitmq.client.*;
import lombok.SneakyThrows;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
* @Author: Zhangbin
* @Date: 2021/5/9 14:55
* @Description:  测试DLX-普通消费者,
 * 以下三种情况，普通队列中的消息将会进入死信队列：
 * 1、普通队列中的消息TTL过期
 * 2、队列达到最大长度
 * 3、消息被拒绝(basic.reject / basic.nack)，并且requeue = false
 *
**/
public class NormalConsumer {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = RabbitConfig.createConnection();
        // 创建消息通道
        Channel channel = connection.createChannel();
        channel.basicQos(0,1,false);//每次拿1条配合72行代码演示消息ttl之后进入死信队列

        //创建一个topic交换机
        channel.exchangeDeclare(RabbitConfig.EXCHANGE_DLX_NONE_NAME, RabbitExchangeTypeEnum.TOPIC.getExchangeType(),false,false,null);
        //创建队列并指定死信进入到哪一个交换机
        Map<String,Object> arguments = new HashMap<String,Object>();
        arguments.put("x-dead-letter-exchange",RabbitConfig.EXCHANGE_DLX_NAME);//设置死信交换机
        //arguments.put("x-message-ttl",5000); // 设置队列的TTL
        arguments.put("x-max-length",5);//设置队列的最大容量
        arguments.put("x-overflow","drop-head");//队列满载之后的溢出规则， drop-head ：(默认,丢弃最先进来的消息接收后面进来的消息)，reject-publish：（直接丢去后面进来的消息）
        channel.queueDeclare(RabbitConfig.QUEUE_DLX_NONE_NAME,false,false,false,arguments);
        //绑定
        channel.queueBind(RabbitConfig.QUEUE_DLX_NONE_NAME,RabbitConfig.EXCHANGE_DLX_NONE_NAME,RabbitConfig.ROUTING_DLX_NONE_KEY);


        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @SneakyThrows
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "UTF-8");

                if(msg.contains("1")){
                    //拒绝并且不重新入队的消息会进入死信队列
                    channel.basicReject(envelope.getDeliveryTag(), false);
                    System.out.println("Received message : " + msg +"--->消息拒收，准备路由到死信队列");
                    return;
                }
                if(msg.contains("2")){
                    //nack不重新入队的消息会进入死信队列
                    System.out.println("Received message : " + msg +"--->消息NACK，准备路由到死信队列");
                    channel.basicNack(envelope.getDeliveryTag(), false, false);
                    return;
                }
                System.out.println("Received message : " + msg);
                Thread.sleep(2000);
                channel.basicAck(envelope.getDeliveryTag(),true);
            }
        };
        channel.basicConsume(RabbitConfig.QUEUE_DLX_NONE_NAME,false,consumer);
    }

}
